package com.bj58.ecdata.hadoop.fourthinfo;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class MSearchCountJob 
{
	public static BufferedReader makeBufferReader(String ifpath) throws Exception 
	{
		//打开读文件
		String charset="utf-8";
		return makeBufferReader(ifpath,charset);
	}
	
	public static BufferedReader makeBufferReader(String ifpath,String charset) throws Exception 
	{
		//打开读文件
		Configuration config = new Configuration();
        FileSystem hdfs = FileSystem.get(config);
        Path path = new Path(ifpath);
        InputStreamReader isr = null;
		BufferedReader fileBR = null;
		
		try
		{
			isr = new InputStreamReader( hdfs.open(path),charset);
			fileBR = new BufferedReader(isr);
			return fileBR;
		}
		catch(Exception e)
		{
			e.printStackTrace();
			System.out.println("open file error!");
			if(fileBR!=null)
			{
				try
				{
					fileBR.close();
				}
				catch(IOException e1)
				{
					e1.printStackTrace();
				}
			}
			if(isr!=null)
			{
				try
				{
					isr.close();
				}
				catch(IOException e1)
				{
					e1.printStackTrace();
				}
			}
			return null;
		}
		
	}
	
	public static String arrayToString(Object[] strArr)
	{
		return arrayToString(strArr,"\t");
	}
	
	public static String arrayToString(Object[] strArr,String sep)
	{
		StringBuilder sb = new StringBuilder();
		for(int i = 0 ; i < strArr.length; i++)
		{
			if(i != 0)
			{
				sb.append(sep);
			}
			sb.append(strArr[i]);
		}
		return sb.toString();
	}
	
	public static Map<String,String> getMapFormHDFSFile(String fpath,String sep) throws Exception
	{
        BufferedReader fileBR = makeBufferReader(fpath);
        
        Map<String,String> lineInfo=new HashMap<String,String>();
		
		//遍历文件
		String strLineBuff="";
		try
		{
			while((strLineBuff=fileBR.readLine()) != null)
			{
				if(strLineBuff!=null && !"".equals(strLineBuff))
				{
					try
					{
						String [] strArr=strLineBuff.split(sep,2);
						lineInfo.put(strArr[0],strArr[1]);
					}
					catch(Exception e)
					{}
				}
			}
		}
		catch(Exception e)
		{
			e.printStackTrace();
			System.out.println(fpath);
		}
		
		killBufferReader(fileBR);
		return lineInfo;
	}
	
	public static void killBufferReader(BufferedReader reader)
	{
		try 
		{
			if(reader!=null)
			{
				reader.close();
			}
		} 
		catch (IOException e) 
		{
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	
	public static final int IP_INDEX=5;
	public static final int REFERRER_URL_INDEX=14;
	public static final int URL_INDEX=3;
	
	
	public static final String iplist="1.192.144.46,1.195.40.4,1.195.40.5,1.56.135.107,1.82.143.49,10.3.15.216,110.109.179.114,110.112.134.87,110.17.68.222,111.182.38.207,112.25.183.33,112.25.183.34,112.25.183.35,112.25.183.36,112.25.183.37,112.25.183.38,112.83.214.150,112.84.69.88,113.204.67.45,113.204.67.46,113.244.10.65,113.3.210.20,113.73.208.23,114.230.127.63,114.238.238.163,115.218.55.9,115.218.57.122,117.22.56.32,117.69.18.30,117.92.143.198,117.92.164.112,118.122.117.44,119.0.118.210,119.114.58.170,119.176.170.106,119.185.89.124,119.191.71.183,119.39.124.198,119.39.124.217,120.192.47.132,120.195.51.179,120.195.51.180,120.210.163.6,120.86.68.240,120.86.68.241,120.86.68.242,120.86.68.243,120.86.68.244,120.86.68.245,120.86.68.246,120.86.68.247,120.86.68.248,120.86.68.249,120.86.68.250,120.86.68.251,120.86.68.252,120.86.68.253,120.86.68.254,120.86.68.255,121.17.184.36,121.62.137.33,122.156.219.194,122.192.93.31,122.227.95.74,122.4.85.170,122.4.85.178,122.96.165.130,123.11.31.158,123.133.247.65,123.152.130.212,123.152.218.208,123.163.133.171,123.171.50.254,123.186.193.144,123.187.31.4,123.52.165.115,123.92.200.148,124.128.220.90,124.128.220.91,124.128.220.92,124.128.220.93,124.128.220.94,124.134.110.72,124.205.149.192,124.205.149.193,124.205.149.194,124.205.149.195,124.205.149.196,124.205.149.197,124.205.149.198,124.205.149.199,124.205.149.200,124.205.149.201,124.205.149.202,124.205.149.203,124.205.149.204,124.205.149.205,124.205.149.206,124.205.149.207,124.205.149.208,124.205.149.209,124.205.149.210,124.205.149.211,124.205.149.212,124.205.149.213,124.205.149.214,124.205.149.215,124.205.149.216,124.205.149.217,124.205.149.218,124.205.149.219,124.205.149.220,124.205.149.221,124.205.149.222,124.205.149.223,124.229.13.188,171.91.151.18,175.1.64.202,175.174.84.29,175.19.190.145,175.19.190.146,175.19.190.147,175.4.81.5,180.119.121.244,180.119.98.60,180.139.138.129,180.139.138.199,183.187.66.116,183.217.183.102,183.62.205.115,183.62.205.126,192.168.0.113,192.168.1.122,192.168.1.202,192.168.2.102,218.25.172.158,218.25.172.65,218.26.104.190,218.56.37.118,218.65.22.27,219.139.133.141,219.150.117.50,220.248.234.194,220.248.235.34,221.1.114.129,221.207.181.224,221.207.181.225,221.207.181.226,221.207.181.227,221.207.181.228,221.207.181.229,221.207.181.230,221.207.181.231,222.185.121.190,222.222.23.118,222.222.23.119,222.222.23.120,222.222.23.121,222.222.23.122,222.222.23.123,222.222.23.124,222.222.23.125,222.222.23.126,222.222.23.34,222.222.23.35,222.245.238.169,222.46.21.53,222.47.60.167,222.47.60.168,222.47.60.169,222.47.60.170,222.47.60.171,222.59.165.106,223.146.5.140,223.198.96.139,223.241.148.146,27.115.13.42,27.115.13.43,27.115.13.44,27.115.13.45,27.115.13.46,27.151.120.177,27.151.120.178,27.17.11.110,27.187.83.220,27.22.38.77,36.42.79.167,42.243.43.162,42.49.153.58,49.71.129.73,58.214.29.7,58.248.12.20,58.248.12.21,58.248.12.22,59.46.65.146,59.46.65.147,59.46.65.148,59.46.65.149,59.52.141.227,60.169.69.154,60.179.199.114,60.19.243.80,60.29.171.42,60.29.171.48,60.29.171.49,60.29.171.50,60.29.171.51,60.29.171.52,60.29.171.53,60.29.171.54,60.29.171.55,60.30.12.0,60.30.12.1,60.30.12.10,60.30.12.11,60.30.12.12,60.30.12.13,60.30.12.14,60.30.12.15,60.30.12.2,60.30.12.3,60.30.12.4,60.30.12.5,60.30.12.6,60.30.12.7,60.30.12.8,60.30.12.9,60.30.22.1,60.30.22.2,61.163.106.34";
	public static final String [] ipVec=iplist.split(",");
	public static final List<String> ipArr=Arrays.asList(ipVec);
	
	
	/**
	 * 用于从url中解析出来pageNum
	 */
	public static final String pageNumReStr="(?:/pn|[&?]page=)(\\d+)";
	public static final Pattern pageNumPattern = Pattern.compile(pageNumReStr);
	
	public static String[] getUrlInfo(String url) throws Exception
	{
		int index=url.indexOf("?");
		String urlPath=(index!=-1)?url.substring(0,index):url;
		String urlParams=(index!=-1)?url.substring(index):"";
		return new String[]{urlPath,urlParams};
	}
	public static String getPageNumFromUrl(String url) throws Exception
	{
		String pageNum=null;
		Matcher matc = pageNumPattern.matcher(url);
		if(!matc.find())
		{
			return "1";
		}
		pageNum = matc.group(1);
		return pageNum;
	}
	
	
	//过滤乱码
	public static boolean regWord(String word){
		// \p{P} 是匹配标点符号
		String reg = "^([=+]|[\\s]|[\\p{P}]|[A-Za-z0-9]|[\u4E00-\u9FA5])+$";
		Pattern p = Pattern.compile(reg);
		Matcher m = p.matcher(word);
		return m.matches();
	}
	public static boolean regUTF(String line){
		String reg = "^(?:[\\x00-\\x7f]|[\\xe0-\\xef][\\x80-\\xbf]{2})+$";
		Pattern p = Pattern.compile(reg);
		Matcher m = p.matcher(line);
		return m.matches();
	}
	
	public static final String  keyWordReStr           =   "(?:/key_|[&?]key=|/jh_)([^?^/^&]+)";
	public static final Pattern keyWordRePatten        =   Pattern.compile(keyWordReStr) ;
	public static String getKeyWordFromUrl(String url) throws Exception {
		Matcher matc = keyWordRePatten.matcher(url);
		if (!matc.find()) {
			return "";
		}
//		return YURLDecode.searchWordDecode(matc.group(1));
		return matc.group(1);
	}
	
	public static Map<String,String> getLineFieldsInfo(String line,
			Map<String,String> cityShInfo,Map<String,String> cateShInfo,
			Map<String,String> cityInfo,Map<String,String> cateInfo,
			Map<String,String> wordTypeInfo) throws Exception
	{
		Map<String,String> lineFieldsInfo=new HashMap<String,String>();
		String [] lineParams=(line!=null)?line.split("\\|"):null;
		if(lineParams==null || lineParams.length<15)
		{
			return null;
		}
		
		
		String url=lineParams[URL_INDEX];
		String ip=lineParams[IP_INDEX];
		
		String uncodeSearchWord=getKeyWordFromUrl(url);
		if(uncodeSearchWord==null || "".equals(uncodeSearchWord))
		{
			return null;
		}
		uncodeSearchWord=uncodeSearchWord.replace("\"","");
		uncodeSearchWord=uncodeSearchWord.replace("'","");
		uncodeSearchWord=uncodeSearchWord.replace("\t"," ").replace("\\","%").trim();
		if(uncodeSearchWord==null || "".equals(uncodeSearchWord) || uncodeSearchWord.length()==0){
			uncodeSearchWord="-";
		}
		
		if(!regWord(uncodeSearchWord)){
			return null;
		}
		
		if(ipArr.contains(ip))
		{
			return null;
		}
		
		String urlPath=(url!=null)?getUrlInfo(url)[0]:"-"; 
		
		String localListName=null;
		
		String cateListName=null;
		
		String localId=null;
		
		String cateId=null;
		
		String t_urlPath=urlPath.replace(".58.com","");
		
		for(String str:t_urlPath.split("/"))
		{
			if(cityShInfo.containsKey(str))
			{
				localListName=str;
				localId=cityShInfo.get(localListName);
			}
			
			if(cateShInfo.containsKey(str))
			{
				cateListName=str;
				if(cateListName!=null && !"".equals(cateListName) && cateListName.length()!=0){
					cateId=cateShInfo.get(cateListName);
				}
				
			}
		}
		
		localId=(localId==null)?"-1":localId;
		
		cateId=(cateId==null)?"-1":cateId;
		
		localListName=(localListName==null)?"unknown":localListName;
		localListName=(localListName.length()>20)?localListName.substring(0,20):localListName;
		
		cateListName=(cateListName==null)?"unknown":cateListName;
		cateListName=(cateListName.length()>20)?cateListName.substring(0,20):cateListName;
		
		
		
		String pageNum=getPageNumFromUrl(url);
		pageNum=pageNum.length()>10?pageNum.substring(0,10):pageNum;
		
		String cityFullPathStr=cityInfo.get(localId);
		String cateFullPathStr=cateInfo.get(cateId);
		cityFullPathStr=(cityFullPathStr!=null)?cityFullPathStr:"\t\t\t";
		cateFullPathStr=(cateFullPathStr!=null)?cateFullPathStr:"\t\t\t\t";
		
		lineFieldsInfo.put("uncodeSearchWord",uncodeSearchWord);
		lineFieldsInfo.put("localListName",localListName);
		lineFieldsInfo.put("cateListName",cateListName);
		lineFieldsInfo.put("localId",localId);
		lineFieldsInfo.put("cateId",cateId);
		lineFieldsInfo.put("pageNum",pageNum);
		lineFieldsInfo.put("cityFullPathStr", cityFullPathStr);
		lineFieldsInfo.put("cateFullPathStr", cateFullPathStr);
		
		return lineFieldsInfo;
	}
	
	
	public static class MyMapper extends Mapper<LongWritable, Text, Text, Text>
	{
		private Map<String,String> cityShInfo=null;
		private Map<String,String> cateShInfo=null;
		private Map<String,String> cityInfo=null;
		private Map<String,String> cateInfo=null;
		public  Map<String,String> wordTypeInfo=null;
		public  Map<String,String> businessInfo=null;
		
		protected void setup(Context context) throws IOException,InterruptedException 
		{
			try 
			{
				String statDate = context.getConfiguration().get("statDate");
				cityShInfo = getMapFormHDFSFile("/dsap/resultdata/ecdataconfig/DispCityListNameInfo/"+ statDate.replace("-", "") + "/part-r-00000",	"\t");
				cateShInfo = getMapFormHDFSFile("/dsap/resultdata/ecdataconfig/DispCateListNameInfo/"+ statDate.replace("-", "") + "/part-r-00000",	"\t");
				
				cityInfo = getMapFormHDFSFile("/dsap/resultdata/ecdataconfig/DispCityInfo/"+ statDate.replace("-", "") + "/part-r-00000",	"\t");
				cateInfo = getMapFormHDFSFile("/dsap/resultdata/ecdataconfig/DispCateInfo/"+ statDate.replace("-", "") + "/part-r-00000",	"\t");
				
				
				businessInfo = getMapFormHDFSFile("/dsap/resultdata/ecdataconfig/DispBusinessTypeInfo/"+ statDate.replace("-", "") + "/part-r-00000",	"\t");
				
			} 
			catch (Exception e) 
			{
			}
		}
		
		
		public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException 
		{
			String line=value.toString();
			try 
			{
				Map<String,String> lineInfo=getLineFieldsInfo(line, cityShInfo, cateShInfo, cityInfo, cateInfo, wordTypeInfo);
				if(lineInfo==null){
					return;
				}
				
				
				String uncodeSearchWord=lineInfo.get("uncodeSearchWord");
				String localListName=lineInfo.get("localListName");
				String cateListName=lineInfo.get("cateListName");
				
				String localId=lineInfo.get("localId");
				String cateId=lineInfo.get("cateId");
				
				String cityFullPathStr=lineInfo.get("cityFullPathStr");
				String cateFullPathStr=lineInfo.get("cateFullPathStr");
				String pageNum=lineInfo.get("pageNum");
				String [] strArr={
						uncodeSearchWord,
						localListName,
						cateListName,
						localId,
						cateId,
						cityFullPathStr,
						cateFullPathStr,
						pageNum
				};
				context.write(new Text(arrayToString(strArr)), new Text("") );
			} 
			catch (Exception e) 
			{
				e.printStackTrace();
			}
		}
	}
	
	public static class MyReducer extends Reducer<Text, Text, NullWritable,Text> 
	{
		public void reduce(Text key, Iterable<Text> values,Context context) throws IOException, InterruptedException 
		{
			int searchCount=0;
			String statDate = context.getConfiguration().get("statDate");
			for (Text v:values) 
			{
				try
				{
					searchCount=searchCount+1;
				}
				catch(Exception e)
				{
					context.write(NullWritable.get(), new Text("reduce while error!!!"+v.toString()) );
					return;
				}
			}
			String val=arrayToString(new String[]{statDate,key.toString(),searchCount+""}, "\t");
			context.write(NullWritable.get(), new Text(val) );
		}
	}
	
	
	// 得到起始范围内的日期列表，包含起始日期，可设置间隔天数
	public static List<String> getDateList(String startDate, String endDate, int intervalDay) 
	{
		List<String> listDate = new ArrayList<String>();

		SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd");
		Date dBegin;
		Date dEnd;
		try 
		{
			dBegin = f.parse(startDate);
			dEnd = f.parse(endDate);
			if (dBegin.getTime() <= dEnd.getTime()) 
			{
				for (long i = dBegin.getTime(); i <= dEnd.getTime(); i += 86400000 * (intervalDay + 1)) 
				{
					Date d = new Date(i);
					String date = f.format(d);
					listDate.add(date);
				}
			} 
			else 
			{
				for (long i = dBegin.getTime(); i >= dEnd.getTime(); i -= 86400000 * (intervalDay + 1)) 
				{
					Date d = new Date(i);
					String date = f.format(d);
					listDate.add(date);
				}
			}
		} 
		catch (ParseException e) 
		{
			e.printStackTrace();
		}
		return listDate;
	}
	public static String getStringDate(Date date, String pattern) {
		try {
			SimpleDateFormat f = new SimpleDateFormat(pattern);
			String strdate = f.format(date);
			return strdate;
		} catch (Exception pe) {
		}
		return null;/**/
	}
	
	public static Date getDateBefore(Date d, int day) 
	{
		Calendar now = Calendar.getInstance();
		now.setTime(d);  
	    now.set(Calendar.DATE, now.get(Calendar.DATE) - day);  
	    return now.getTime(); 
	}  
	public static Date getDateAfter(Date d, int day) {
		Calendar now = Calendar.getInstance();  
		now.setTime(d);  
		now.set(Calendar.DATE, now.get(Calendar.DATE) + day);  
		return now.getTime();  
	} 
	
	public static int getArgsParamsNum(String[] args){
		if(args==null || args.length==0){
			return 0;
		}else{
			return args.length;
		}
	}
	
	public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException 
	{
		//job执行参数设置
		String JOB_NAME="MSearchCountJob";
		int REDUCE_TASK_NUM=10;
		String [] IPATH_ARR={
				"/dsap/rawdata/m.track.58.com/yyyyMMdd"
		};
		String OPATH=String.format("/dsap/resultdata/ysa/userdata/%s/yyyyMMdd",JOB_NAME);
		
		// 字符串日期遍历
		Configuration configuration = new Configuration();
		String[] otherArgs = null;
		try
		{
			otherArgs = new GenericOptionsParser(configuration, args).getRemainingArgs();
		} 
		catch (IOException e) 
		{
			e.printStackTrace();
		}
		
		int argsNum=getArgsParamsNum(otherArgs);
		
		List<String> dateList=null;
		String yestoday=getStringDate(getDateBefore(new Date(),1), "yyyy-MM-dd");
		if(argsNum==0){
			dateList=getDateList(yestoday, yestoday, 0);
		}else if(argsNum==1){
			dateList=getDateList(otherArgs[0], otherArgs[0], 0);
		}else if(argsNum==2){
			dateList=getDateList(otherArgs[0], otherArgs[1], 0);
		}
		for (String statDate : dateList) 
		{
			System.out.println(statDate);
			try
			{
				configuration.set("statDate", statDate);
				Job job = new Job(configuration, JOB_NAME);
				job.setJarByClass(MSearchCountJob.class);
				job.setNumReduceTasks(REDUCE_TASK_NUM);
				job.setMapperClass(MSearchCountJob.MyMapper.class);
				job.setReducerClass(MSearchCountJob.MyReducer.class);
				job.setMapOutputKeyClass(Text.class);
				job.setMapOutputValueClass(Text.class);
				job.setOutputKeyClass(NullWritable.class);
				job.setOutputValueClass(Text.class);
				
				FileSystem hdfs = FileSystem.get(configuration);
				if(hdfs.exists(new Path(OPATH.replace("yyyyMMdd", statDate.replace("-","")))))
				{
					hdfs.delete(new Path(OPATH.replace("yyyyMMdd", statDate.replace("-",""))), true);
				}
				
				for(String ifPath:IPATH_ARR)
				{
					FileInputFormat.addInputPath(job,new Path(ifPath.replace("yyyyMMdd", statDate.replace("-",""))));	
				}
				FileOutputFormat.setOutputPath(job, new Path(OPATH.replace("yyyyMMdd", statDate.replace("-","")))); 
				System.out.println(job.waitForCompletion(true) ? 0 : 1);
			} 
			catch (Exception e) 
			{
				e.printStackTrace();
			}
		}
	}
}



